package com.da.javatest.interview.session2;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

import org.apache.commons.lang3.StringUtils;

/**
 * @author chenlida
 * @date 2020/3/31 22:24
 * @description 使用阻塞队列的生产者消费者模式
 */
public class Test19PCBlockQueue {
    private volatile boolean FLAG = true;
    private AtomicInteger ai = new AtomicInteger();
    BlockingQueue<String> blockingQueue;

    public Test19PCBlockQueue(BlockingQueue<String> blockingQueue) {
        this.blockingQueue = blockingQueue;
    }

    public void producter() throws Exception {
        boolean retValue;
        String data;
        while (FLAG) {
            data = ai.getAndIncrement() + "";
            retValue = blockingQueue.offer(data, 2, TimeUnit.SECONDS);
            if (retValue) {
                System.out.println(Thread.currentThread().getName() + "产生数据：" + data + " 成功");
            } else {
                System.out.println(Thread.currentThread().getName() + "产生数据：" + data + " 失败");
            }
            TimeUnit.SECONDS.sleep(1);
        }
        System.out.println("生产停止。。。");
    }

    public void consumer() throws Exception {
        String result;
        while (FLAG) {
            result = blockingQueue.poll(2, TimeUnit.SECONDS);
            if (StringUtils.isBlank(result)) {
                FLAG = false;
                System.out.println(Thread.currentThread().getName() + "超过2s没获取数据，消费退出");
                System.out.println();
                return;
            }
            System.out.println(Thread.currentThread().getName() + "消费数据：" + result);
        }
    }

    public void stop() {
        FLAG = false;
    }

    public static void main(String[] args) throws InterruptedException {
        Test19PCBlockQueue t = new Test19PCBlockQueue(new ArrayBlockingQueue<>(10));
        new Thread(() -> {
            try {
                t.producter();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }, "生产者线程").start();

        new Thread(() -> {
            try {
                t.consumer();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }, "消费者线程").start();

        TimeUnit.SECONDS.sleep(6);
        t.stop();
    }
}
